package com.example.wzh.rxdemo.rx;

import androidx.annotation.NonNull;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

final class ObservableRepeatAccordingResult<T> extends Observable<T> {
    private final ObservableSource<T> source;
    private final Predicate<T> predicate;

    public ObservableRepeatAccordingResult(Observable<T> source, Predicate<T> predicate) {
        this.source = source;
        this.predicate = predicate;
    }

    @Override
    public void subscribeActual(Observer<? super T> observer) {
        Disposable disposable = Disposable.empty();
        observer.onSubscribe(disposable);
        RepeatAccordingResultObserver<T> rs = new RepeatAccordingResultObserver<>(observer, disposable, source, predicate);
        rs.subscribeNext();
    }

    static final class RepeatAccordingResultObserver<T> extends AtomicInteger implements Observer<T> {

        private static final long serialVersionUID = -7098360935104053232L;

        final Observer<? super T> downstream;
        Disposable disposable, sourceDisposable;
        final ObservableSource<? extends T> source;
        final Predicate<T> predicate;

        RepeatAccordingResultObserver(Observer<? super T> actual, Disposable disposable, ObservableSource<? extends T> source, Predicate<T> predicate) {
            this.downstream = actual;
            this.disposable = disposable;
            this.source = source;
            this.predicate = predicate;
        }

        @Override
        public synchronized void onSubscribe(@NonNull Disposable d) {
            sourceDisposable = d;
        }

        @Override
        public void onNext(@NonNull T t) {
            if (disposable.isDisposed()) {
                return;
            }
            downstream.onNext(t);
            if (predicate.test(t)) {
                sourceDisposable.dispose();
                subscribeNext();
            }
        }

        @Override
        public void onError(@NonNull Throwable t) {
            sourceDisposable.dispose();
            if (disposable.isDisposed()) {
                return;
            }
            downstream.onError(t);
        }

        @Override
        public void onComplete() {
            sourceDisposable.dispose();
            if (disposable.isDisposed()) {
                return;
            }
            downstream.onComplete();
        }

        synchronized void subscribeNext() {
            if (disposable.isDisposed()) {
                return;
            }
            source.subscribe(this);
        }
    }
}
